package com.sjtu.zookeeper;

import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.x.discovery.*;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.apache.curator.x.discovery.strategies.RandomStrategy;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.commons.util.InetUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.concurrent.ConcurrentMap;

/**
 * Spring只能托管自己在zookeeper组件里声明的serviceDiscovery，如果再声明一个的话会因为和@ConditionOnMissingBean冲突导致报错，
 * 故这里就采用静态注入的方式声明一个自用的对象
 * nerine
 * 2/5/2020
 */
@Slf4j
@Component
public class ServiceDiscoverUtil implements InitializingBean {

    private static CuratorFramework curatorFramework;

    private static ServiceDiscovery<InstanceDetails> sjtuServiceDiscovery;

    private static final ConcurrentMap<String, ServiceProvider<InstanceDetails>> INSTANCES = Maps.newConcurrentMap();

    @Autowired
    private ZookeeperProperties properties;
    @Autowired
    private ApplicationContext context;
    @Autowired
    private InetUtils inetUtils;

    private static final String DEFAULT_URI_SPEC = "{scheme}://{address}:{port}";

    public static ServiceProvider<InstanceDetails> getServiceProvider(String systemId) {
        ServiceProvider<InstanceDetails> provider = INSTANCES.get(systemId);
        if(provider == null) {
            provider = sjtuServiceDiscovery.serviceProviderBuilder().serviceName(systemId)
                    .providerStrategy(new RandomStrategy<>()).build();
            try {
                provider.start();
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                CloseableUtils.closeQuietly(provider);
            }
            INSTANCES.put(systemId, provider);
        }
        return provider;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if(properties.isReuseExistedCurator()) {
            curatorFramework = context.getBean(CuratorFramework.class);
        }else {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(properties.getBaseSleepTimeMs(),
                    properties.getMaxRetries(), properties.getMaxSleepMs());

            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
            builder.connectString(properties.getConnectString());
            curatorFramework = builder.retryPolicy(retryPolicy).build();
            curatorFramework.start();
            log.trace("blocking until connected to zookeeper for "
                    + properties.getBlockUntilConnectedWait()
                    + properties.getBlockUntilConnectedUnit());
            curatorFramework.blockUntilConnected(properties.getBlockUntilConnectedWait(),
                    properties.getBlockUntilConnectedUnit());
            log.trace("connected to zookeeper");
        }

        JsonInstanceSerializer<InstanceDetails> serializer = new JsonInstanceSerializer<>(InstanceDetails.class);
        sjtuServiceDiscovery = ServiceDiscoveryBuilder.builder(InstanceDetails.class)
                .client(curatorFramework).basePath(InstanceDetails.BASE_PATH).serializer(serializer).build();
        sjtuServiceDiscovery.start();
        registerService();
    }

    /**
     * 因payload类不同，如果直接读取services节点的信息的话会类型转换报错，这里需要重复把
     * service的注册信息再通过相同的payload添加一份到clients节点去
     */
    private void registerService() throws Exception {
        String appName = context.getEnvironment().getProperty("sjtu.application.name",
                "application");
        int port = Integer.parseInt(context.getEnvironment().getProperty("server.port"));

        InetUtils.HostInfo hostInfo = inetUtils.findFirstNonLoopbackHostInfo();
        String instanceHost = hostInfo.getHostname();
        String instanceIpAddress = hostInfo.getIpAddress();


        InstanceDetails zookeeperInstance = new InstanceDetails(context.getId(),
                appName, Collections.emptyMap());
        ServiceInstance<InstanceDetails> serviceInstance = ServiceInstance.<InstanceDetails>builder()
                .name(appName)
                .address(instanceIpAddress)
                .port(port)
                .payload(zookeeperInstance)
                .uriSpec(new UriSpec(DEFAULT_URI_SPEC))
                .build();
        sjtuServiceDiscovery.registerService(serviceInstance);
    }
}
